package io.openmessaging;

import io.openmessaging.map.TopicQueueMap;
import io.openmessaging.storage.PersistentMemoryStorageFileChannel;
import io.openmessaging.storage.SSDStorage;
import io.openmessaging.storage.Storage;
import io.openmessaging.test.TestSSD;
import io.openmessaging.utils.Log;
import io.openmessaging.utils.Statistic;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 这是一个简单的基于内存的实现，以方便选手理解题意；
 * 实际提交时，请维持包名和类名不变，把方法实现修改为自己的内容；
 */
public class DefaultMessageQueueImpl extends MessageQueue {

    {
        //定时退出
        Thread exit = new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(600);
            } catch (Throwable e) {
                e.printStackTrace();
            }
            System.exit(0);
        });
        exit.start();

//        AdderTest.main(null);
//        TestSSD.main(null);
//        TestSSDAppend.main(null);
//        TestSSDPmemAppend.main(null);
//        TestPmem.test();

        Log.i("start");
        new File(Conf.ssdPath).mkdirs();
        Statistic.startLog();
    }

    private PersistentTopicMap topicMap = new PersistentTopicMap();
    //每个 queue 的下一个 offset
    TopicQueueMap<long[]> queuesNextOffset = new TopicQueueMap<>(long[].class);

    private Storage persistentMemoryStorage = new PersistentMemoryStorageFileChannel();
    private SSDStorage ssdStorage = new SSDStorage();
//    private SSDStorage4 ssdStorage = new SSDStorage4();
//    private SSDStorage2 ssdStorage = new SSDStorage2();
//    private SSDStorage3 ssdStorage = new SSDStorage3();
//    private SSDStorage33 ssdStorage = new SSDStorage33();
//    private SSDStorageNoQueue ssdStorage = new SSDStorageNoQueue();

    @Override
    public long append(String topicName, int queueId, ByteBuffer data) {
        Statistic.Append.add(data.remaining());

        Topic topic = topicMap.getOrCreate(topicName);
        long[] offset = queuesNextOffset.computeIfAbsent(topic.id, queueId, () -> new long[1]);
        ByteBuffer directBuffer = ByteBuffer.allocateDirect(data.remaining());
        directBuffer.put(data);
        directBuffer.flip();
        Message m = new Message(topic, queueId, offset[0]++, directBuffer);
        //        if (m.topic.name.equals("topic22") && m.queueId == 937) {
//            Log.i("append start ", m.topic.name, "_", m.topic.id, "_", m.queueId, "_", m.offset, "_", m.data.remaining(), "__", m.data.get(m.data.position()));
//        }
        persistentMemoryStorage.save(m);
        ssdStorage.save(m);
//        if (m.topic.name.equals("topic22") && m.queueId == 937) {
//            Log.i("append end ", m.topic.name, "_", m.topic.id, "_", m.queueId, "_", m.offset, "_", m.data.remaining(), "__", m.data.get(m.data.position()));
//        }
        return m.offset;
    }

    @Override
    public Map<Integer, ByteBuffer> getRange(String topic, int queueId, long offset, int fetchNum) {
//        if (topic.equals("topic22") && queueId == 937) {
//            Log.i("getRange start ", topic, "_", queueId, "_", offset, "_", fetchNum);
//        }
        Topic t = topicMap.get(topic);
        if (t == null)
            return Collections.EMPTY_MAP;

        Map<Integer, ByteBuffer> ret = new HashMap<>();
        for (int i = 0; i < fetchNum; i++) {
            ByteBuffer data = ssdStorage.get(t, queueId, offset + i, persistentMemoryStorage);
            if (data == null)//队列结尾
                break;
            ret.put(i, data);
//            if (topic.equals("topic22") && queueId == 937) {
//                Log.i("get ", t.name, "_", t.id, "_", queueId, "_", offset + i, "_", data.remaining(), "__", data.get(data.position()));
//            }
        }
//        if (topic.equals("topic22") && queueId == 937) {
//            Log.i("getRange end ", topic, "_", queueId, "_", offset, "_", fetchNum, "return size = ", ret.size());
//        }
        return ret;
    }
}
